﻿using Misakai.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.NetworkInformation;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Reflexive.IO
{
    public static class Extensions
    {
        // Misakai.Kafka.BinaryHelper
        public static int ToInt32(this byte[] value)
        {
            return BitConverter.ToInt32(value.Reverse<byte>().ToArray<byte>(), 0);
        }

        public static TcpState GetState(this TcpClient tcpClient)
        {
            var connection = IPGlobalProperties.GetIPGlobalProperties()
              .GetActiveTcpConnections()
              .SingleOrDefault(x => x.LocalEndPoint.Equals(tcpClient.Client.LocalEndPoint));
            return connection != null ? connection.State : TcpState.Unknown;
        }

        public static IPAddress TryConvertToIPAddress(this string value)
        {
            IPAddress ip = IPAddress.None;
            if (!String.IsNullOrEmpty(value) && !IPAddress.TryParse(value, out ip))
            {
                return Dns.GetHostAddresses(value).First();
            }
            return ip;
        }

        public static void OnNext<T>(this SynchronizedCollection<Subscription<T>> subscribers, T value) where T : class
        {
            List<Exception> list = new List<Exception>();
            foreach (var observer in subscribers)
            {
                try
                {
                    observer.OnNext(value);
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.TraceError(e.ToString());
                    list.Add(e);
                }
            }
            if(list.Count() > 0)
                throw new AggregateException(list.ToArray());
        }

        public static void OnError<T>(this SynchronizedCollection<Subscription<T>> subscribers, Exception e) where T : class
        {
            List<Exception> list = new List<Exception>();
            foreach (var observer in subscribers)
            {
                try
                {
                    observer.OnError(e);
                }
                catch (Exception ex)
                {
                    System.Diagnostics.Trace.TraceError(e.ToString());
                    list.Add(ex);
                }
            };
            if (list.Count() > 0)
                throw new AggregateException(list.ToArray());

        }

        public static void OnCompleted<T>(this SynchronizedCollection<Subscription<T>> subscribers) where T : class
        {
            List<Exception> list = new List<Exception>();
            foreach (var observer in subscribers)
            {
                try
                {
                    observer.OnCompleted();
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.TraceError(e.ToString());
                    list.Add(e);
                }
            }
            if (list.Count() > 0)
                throw new AggregateException(list.ToArray());
        }

        public static async Task<T> WithCancellation<T>(this Task<T> task, CancellationToken cancellationToken)
        {
            Action<object> action = null;
            TaskCompletionSource<bool> taskCompletionSource = new TaskCompletionSource<bool>();
            if (action == null)
            {
                action = delegate(object source)
                {
                    ((TaskCompletionSource<bool>)source).TrySetResult(true);
                };
            }
            using (cancellationToken.Register(action, taskCompletionSource))
            {
                if (task != await Task.WhenAny(new Task[]
		            {
			            task,
			            taskCompletionSource.Task
		            }))
                {
                    throw new OperationCanceledException(cancellationToken);
                }
            }
            return await task;
        }

        public static RetryableTask<KafkaOptions> GetOptionsAsRetryable(this KafkaBase obj, string endpoint, CancellationToken token,  TimeSpan interval, int? count = null)
        {
            return new RetryableTask<KafkaOptions>(() => obj.GetOptions(endpoint, Timeout.InfiniteTimeSpan), token, interval, count);
        }
    }
}
